package com.aotain.nyx.abnormal;

import org.apache.commons.codec.binary.Base64;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.log4j.Logger;

public class AbnormalStatMap implements MapFunction<Tuple2<String,String>, Tuple2<String,AbnormalStatTuple>> {

	/** 
	* @Fields serialVersionUID : TODO(用一句话描述这个变量表示什么) 
	*/ 
	private static final long serialVersionUID = 8930209552904219414L;
	
	
	@Override
	public Tuple2<String, AbnormalStatTuple> map(Tuple2<String, String> value)
			throws Exception {
		// TODO Auto-generated method stub
		Tuple2<String,AbnormalStatTuple> ret = new Tuple2<String,AbnormalStatTuple>();
		String[] items = value.f1.split("\\|",-1);
//		Logger.getLogger(getClass()).info("#####AbnormalStatMap#### tuple is " + value.f1);
		if(items.length <= 10)
			return ret;
		String sip = items[2];
		String dip = items[0];
		String gis = new String(Base64.decodeBase64(items[4]));
		String idc = items[9];
		String type = items[7];
		String dport = items[1];
		String sport = items[3];
		String flowdir = items[19];
		long delayTime = Long.parseLong(items[8]);
		long upStreamOctets = 0; //UPSTREAMOCTETS
		long upStreamPacket = 0; //UPSTREAMPACKET			
		upStreamOctets = Long.parseLong(items[10]);
		upStreamPacket = Long.parseLong(items[11]);
		
		AbnormalStatTuple stat = new AbnormalStatTuple();
		stat.setSourceIP(sip);
		stat.setDestIP(dip);
		stat.setGIS(gis);
		stat.setType(type);
		stat.setDestPort(dport);
		stat.setSourcePort(sport);
		stat.setDelayTime(delayTime);
		stat.setUpStreamOctets(upStreamOctets);
		stat.setUpStreamPacket(upStreamPacket);
		stat.setIDC(idc);
		stat.setFlowDirection(flowdir);

//		String key = dip + "*" + idc + "*" + sip + "*" + gis + "*" + type + "*" + dport ;
		String key = dip + "*" + idc+"*"+dport; ;
		
//		Logger.getLogger(getClass()).info("#####AbnormalStatMap####  tuple  " + stat.toString());
		ret.setFields(key, stat);
//		System.out.println("$$$$$$AbnormalStatMap$$$$$$ "+System.currentTimeMillis());
		return ret;
		
	}

}
